package com.cn.broker;

import com.cn.coder.MessageDecoder;
import com.cn.coder.MessageEncoder;
import com.cn.common.Pair;
import com.cn.common.processor.INettyRequestProcessor;
import com.cn.msg.Message;
import com.cn.remote.AbStractNetty;
import com.cn.remote.ChannelListener;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.*;
import io.netty.channel.epoll.Epoll;
import io.netty.channel.epoll.EpollEventLoopGroup;
import io.netty.channel.epoll.EpollServerSocketChannel;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.concurrent.DefaultEventExecutorGroup;

import java.net.InetSocketAddress;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

public class BrokerServer extends AbStractNetty {

    private int port;

    private String address;

    private EventLoopGroup boss;

    private EventLoopGroup worker;

    private ServerBootstrap bootstrap;

    private BrokerController brokerController;

    private ExecutorService defaultExecutorService;

    private ChannelListener channelListener;

    public void registerChannelListener(ChannelListener channelListener) {
        this.channelListener = channelListener;
    }

    public BrokerServer(int port, String address, BrokerController brokerController) {
        super(500);
        this.callbackExecutor = Executors.newFixedThreadPool(16);

        this.brokerController = brokerController;
        this.port = port;
        this.address = address;

        this.bootstrap = new ServerBootstrap();
        this.defaultExecutorService = Executors.newCachedThreadPool();

        boolean linux = Epoll.isAvailable();
        if (linux) {
            this.boss = new EpollEventLoopGroup(1, new ThreadFactory() {
                private AtomicInteger atomicInteger = new AtomicInteger(0);

                @Override
                public Thread newThread(Runnable r) {
                    return new Thread(r, String.format("boss线程：%d", this.atomicInteger.incrementAndGet()));
                }
            });
        } else {
            this.boss = new NioEventLoopGroup(1, new ThreadFactory() {
                private AtomicInteger atomicInteger = new AtomicInteger(0);

                @Override
                public Thread newThread(Runnable r) {
                    return new Thread(r, String.format("boss线程：%d", this.atomicInteger.incrementAndGet()));
                }
            });
        }


        //工作线程
        if (linux) {
            this.worker = new EpollEventLoopGroup(4, new ThreadFactory() {
                private AtomicInteger atomicInteger = new AtomicInteger(0);

                @Override
                public Thread newThread(Runnable r) {
                    return new Thread(r, String.format("worker线程：%d", this.atomicInteger.incrementAndGet()));
                }
            });
        } else {
            this.worker = new NioEventLoopGroup(4, new ThreadFactory() {
                private AtomicInteger atomicInteger = new AtomicInteger(0);

                @Override
                public Thread newThread(Runnable r) {
                    return new Thread(r, String.format("worker线程：%d", this.atomicInteger.incrementAndGet()));
                }
            });
        }


        DefaultEventExecutorGroup defaultEventExecutorGroup = new DefaultEventExecutorGroup(4, new ThreadFactory() {
            private AtomicInteger atomicInteger = new AtomicInteger(0);

            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r, String.format("ServerCodecThread_%d", this.atomicInteger.incrementAndGet()));
            }
        });

        this.bootstrap.group(boss, worker);
        if (linux == false) {
            bootstrap.channel(NioServerSocketChannel.class);//linux用 EpollServerSocketChannel ；window用NioServerSocketChannel
        } else {
            bootstrap.channel(EpollServerSocketChannel.class);//linux用 EpollServerSocketChannel ；window用NioServerSocketChannel
        }

        this.bootstrap.option(ChannelOption.SO_BACKLOG, 1024)
                .option(ChannelOption.SO_REUSEADDR, true)
                .option(ChannelOption.SO_KEEPALIVE, false)
                .childOption(ChannelOption.TCP_NODELAY, true)
                .childOption(ChannelOption.SO_SNDBUF, 65535)// 发送缓冲大小
                .childOption(ChannelOption.SO_RCVBUF, 65535);// 接收缓冲大小

        this.bootstrap.childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);//worker线程内存池配置
        this.bootstrap.childHandler(new BrokerServerHandler(defaultEventExecutorGroup));
        //该行代码用来检测netty内存泄漏
//        ResourceLeakDetector.setLevel(ResourceLeakDetector.Level.ADVANCED);

//        System.setProperty("io.netty.leakDetection.maxRecords", "100");
//        System.setProperty("io.netty.leakDetection.acquireAndReleaseOnly", "true");
//        ResourceLeakDetector.setLevel(ResourceLeakDetector.Level.PARANOID);

    }


    public void start() throws InterruptedException {
        if (this.bootstrap == null) throw new RuntimeException("初始化broker失败");
        ChannelFuture channel = bootstrap.bind(address, port).sync();
        InetSocketAddress addr = (InetSocketAddress) channel.channel().localAddress();
        this.port = addr.getPort();
        System.out.println("broker服务已启动：" + addr.getAddress().getHostAddress() + ":" + addr.getPort());
    }


    public void stop() {
        if (this.boss != null) {
            this.boss.shutdownGracefully();
            this.worker.shutdownGracefully();
        }
    }

    public void registerProcessor(int code, INettyRequestProcessor processor, ExecutorService executorService) {
        if (executorService == null) {
            executorService = defaultExecutorService;
        }
        Pair<INettyRequestProcessor, ExecutorService> pair = new Pair<>(processor, executorService);
        this.processTables.put(code, pair);
    }

    class BrokerServerHandler extends ChannelInitializer<SocketChannel> {


        private DefaultEventExecutorGroup defaultEventExecutorGroup;

        public BrokerServerHandler(DefaultEventExecutorGroup defaultEventExecutorGroup) {
            this.defaultEventExecutorGroup = defaultEventExecutorGroup;
        }

        @Override
        protected void initChannel(SocketChannel ch) throws Exception {
            //如果消息是自定义类型，那么用以下处理器
            //如果是自定义消息，那么用以下代码
            ch.pipeline().addLast(this.defaultEventExecutorGroup,
                    new MessageEncoder(),//编码器
                    new MessageDecoder(),//解码器
                    new IdleStateHandler(0, 0, 6, TimeUnit.SECONDS),//心跳检测，不管读写，只要超过6秒钟，就主动释放客户端占用的资源
                    new HeartHandler(),
                    new ServerHandler()
            );
        }
    }

    class ServerHandler extends SimpleChannelInboundHandler<Message> {

        public ServerHandler() {
        }

        @Override
        protected void channelRead0(ChannelHandlerContext ctx, Message message) throws Exception {
            process(message, ctx);
        }

        @Override
        public void channelInactive(ChannelHandlerContext ctx) throws Exception {
            ctx.close();
            channelListener.unregisterEvent(ctx);
        }

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            ctx.close();
            channelListener.unregisterEvent(ctx);
        }
    }

    class HeartHandler extends ChannelDuplexHandler {
        @Override
        public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
            if (evt instanceof IdleStateEvent) {
                IdleState state = ((IdleStateEvent) evt).state();
                if (state.equals(IdleState.ALL_IDLE)) {
                    //关闭通道
                    ctx.channel().close().addListener(new ChannelFutureListener() {
                        @Override
                        public void operationComplete(ChannelFuture channelFuture) throws Exception {
                            System.out.println("与失去联系的客户端通道正确关闭！");
                        }
                    });
                }
            }
            ctx.fireUserEventTriggered(evt);
        }

        @Override
        public void disconnect(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {
            ctx.close();
        }

        @Override
        public void close(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {
            ctx.close();
        }

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            ctx.close();
//            cause.printStackTrace();

        }

        @Override
        public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
            super.channelRegistered(ctx);
        }

        @Override
        public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
            super.channelUnregistered(ctx);
        }

        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            super.channelActive(ctx);
        }

        @Override
        public void channelInactive(ChannelHandlerContext ctx) throws Exception {
            super.channelInactive(ctx);
        }
    }

}
